package org.hbhk.hms.consumer;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.hbhk.hms.entity.HmsShard;
import org.hbhk.hms.entity.HmsTopic;
import org.hbhk.hms.mapper.HmsTopicMapper;
import org.hbhk.hms.mq.MqConsumer;
import org.hbhk.hms.mq.MqSender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;

/**
 * 
 * 消费处理，对应的topic未触发数据
 *
 */
@Service
@Order(2)
public class HmsTopicConsumer implements InitializingBean {

	private Logger log = LoggerFactory.getLogger(getClass());

	@Value("${mq.subscribe.HmsDelay:HmsDelay}")
	private String subscribe = "HmsDelay";

	@Value("${mq.reconsumeTimes:3}")
	public static Integer reconsumeTimes = 3;
	
	
	@Value("${mq.bizBatchNum:300}")
	public  Integer bizBatchNum = 300;

	@Autowired
	private HmsTopicMapper hmsTopicMapper;

	private DefaultMQPushConsumer consumer;

	@Autowired
	private MqConsumer mqConsumer;

	public void init() throws Exception {

		log.error("HmsTopic consumer start...");

		consumer = mqConsumer.get("HmsDelay_consumer");
		consumer.subscribe(subscribe, "*");
		consumer.registerMessageListener(new MessageListenerConcurrently() {

			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
				MessageExt me = msgs.get(0);
				try {
					String topic = me.getTopic();
					String tags = me.getTags();
					String keys = me.getKeys();
					String msgBody = new String(me.getBody());
					log.debug("topic: " + topic + ",tags: " + tags + ",key: " + keys + ",msgBody: " + msgBody);
					// 从数据库查询当前需要触发的数据,到对应的mq,待业务系统消费处理
					HmsShard shard = JSON.parseObject(msgBody, HmsShard.class);
					Long cDate = shard.getCreateTime().getTime();
					QueryWrapper<HmsTopic> queryWrapper = new QueryWrapper<>();
					queryWrapper.eq("status", 1);
					queryWrapper.eq("target_topic", shard.getShard());
					queryWrapper.le("delay", cDate);
					Page<HmsTopic> pageSize = new Page<HmsTopic>(0, 10000, false);
					Page<HmsTopic> hmsTopicListPage = hmsTopicMapper.selectPage(pageSize, queryWrapper);

					if (hmsTopicListPage == null) {
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

					}
					List<HmsTopic> hmsTopicList = hmsTopicListPage.getRecords();

					if (hmsTopicList == null || hmsTopicList.size() == 0) {
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
					int num = bizBatchNum;
					for (int j = 0, size = hmsTopicList.size(); j < size; j += num) {
						int toIndex = j + num > size ? size : j + num;
						List<HmsTopic> subList = hmsTopicList.subList(j, toIndex);
						MqSender.sendMessage(subList);
						List<Long> ids = Lists.newArrayList();
						for (HmsTopic hmsTopic : subList) {
							ids.add(hmsTopic.getId());
						}
						HmsTopic hmsTopic = new HmsTopic();
						hmsTopic.setStatus(2);
						QueryWrapper<HmsTopic> updateWrapper = new QueryWrapper<>();
						updateWrapper.in("id", ids);
						updateWrapper.eq("target_topic", shard.getShard());
						hmsTopicMapper.update(hmsTopic, updateWrapper);

					}

				} catch (Exception e) {
					log.error(e.getMessage(), e);
					int reconsumeTimes = me.getReconsumeTimes();
					log.error("reconsumeTimes: " + reconsumeTimes);
					if (reconsumeTimes >= HmsTopicConsumer.reconsumeTimes) {
						return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
					}
					return ConsumeConcurrentlyStatus.RECONSUME_LATER;
				}
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		});

		consumer.start();
		log.error("HmsTopic consumer end...");

	}

	@Override
	public void afterPropertiesSet() throws Exception {
		init();

	}
}
